package cn.edu.thu.laud.thrift.service;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.SchemaDisagreementException;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.edu.thu.laud.com.utils.IOUtils;
import cn.edu.thu.laud.jdbc.util.LaUDJdbcUtils;
import cn.edu.thu.laud.thrift.LaUDDaemon;
import cn.edu.thu.laud.thrift.LaUDDataServer;
import cn.edu.thu.laud.thrift.LaUDLogUtils;
import cn.edu.thu.laud.thrift.service.ServiceFileInfo.FileStoreType;


public class DataTransferService implements Runnable {
	//ConcurrentHashMap<String, LaUDClientState> authorizedUsers;
	//InputStream in;
	//OutputStream out;
	public static String HDFS_PATH=LaUDLogUtils.parser.getElementString("HDFSDataStoreURL", "hdfs://localhost:9000");
	public static String CASSANDRA_PATH=LaUDLogUtils.parser.getElementString("CassandraDataStoreURL", "cfs://localhost:9000");

	static Logger logger = LoggerFactory.getLogger(DataTransferService.class);
	Socket socket;
	int timeout=60*3*1000;
	LaUDClientState state=null;
	FileSystem fs=null;
	FileStoreType fsType=FileStoreType.HDFS;
	Configuration conf = new Configuration();
	public DataTransferService( Socket socket) throws IOException {
		this.socket=socket;
		socket.setSoTimeout(timeout);
	}
	ServiceFileInfo info;
	public  boolean readFolder( Path path,FileSystem fs,DataOutputStream out,int length) throws IOException //throws IOException
	{
		FSDataInputStream in=null;
		boolean result=true;
		FileStatus[] fileStatusArray=null;
		fileStatusArray = fs.listStatus(path);
		for (FileStatus fileStatus : fileStatusArray) {
			Path onepath=fileStatus.getPath();
			if (fs.isFile(onepath)) {
				logger.debug(fileStatus.getLen()+"");
				out.writeLong(fileStatus.getLen());
				logger.debug(onepath.toString().substring(length));
				out.writeUTF(onepath.toString().substring(length));
				try{
					in=fs.open(onepath);
					IOUtils.copyStream(in,out,4096,fileStatus.getLen(),null);
				}
				finally{
					IOUtils.close(in);
				}
			} else if ((fileStatus.isDir())) {
				result=readFolder(fileStatus.getPath(),fs,out,length);
			}
		}
		return result;
	}

	//getkeyspace error:throw
	//delete --error: nothing
	//readLong filesize  
	////readUTF  filepath
	////readBytes 
	//write true
	private boolean writeIntoLaud(DataInputStream inData,DataOutputStream outData) throws InvalidRequestException, IOException, UnavailableException, TimedOutException, SchemaDisagreementException {
		String keyspace=state.getKeyspace();
		//close old fs first.
		if(info.isOverwrite()){
			try {

				fs.delete(new Path(info.getLocation()), true);
			} catch (IOException e2) {
				e2.printStackTrace();
				logger.debug(e2.getMessage());
			}
		}
		long fileSize=-1;
		OutputStream out=null;
		try {
			while((fileSize=inData.readLong())!=-1){						
				String filePath=null;
				filePath = inData.readUTF();
				filePath=info.location+filePath;
				out=fs.create(new Path(filePath), info.isOverwrite());
				IOUtils.copyStream(inData, out, 4096,fileSize, null);
			}
		} finally{
			IOUtils.close(out);
		}

		boolean insert=LaUDDaemon.instance.getCassandraServer().file_internal_insert(state, info.keyBuffer, info.columnFamily, info.columnNameBuffer,info.columnValueBuffer, info.attrs);
		//		LaUDUpdateStatement updateStatement=new LaUDUpdateStatement(keyspace, info.columnFamily, info.keyName, Collections.singletonList(info.columnName), Collections.singletonList(info.columnValue), Collections.singletonList(info.key), info.attrs);
		//		try {
		//			LaUDQueryProcessor.processStatement(new CQLStatement(StatementType.UPDATE, updateStatement), state);
		//		} catch (RecognitionException e) {
		//			e.printStackTrace();
		//			logger.info(e.getMessage());
		//			outData.writeBoolean(false);
		//		}
		outData.writeBoolean(insert);
		return true;
	}
	//write true(mutil files)  false(sigle file)
	//wirte length
	private boolean readFromLaud(DataInputStream inData,DataOutputStream outData) throws IOException{
		Path path=new Path(info.location);
		FileStatus fstatus=null;

		boolean isexist=fs.exists(path);
		if(isexist){
			fstatus = fs.getFileStatus(path);
		}else{
			throw new IOException("files not found.");
		}

		int length=fs.getUri().toString().length();
		boolean readResult=false;
		if(fstatus.isDir()){

			outData.writeBoolean(true);//mutli files

			readResult=readFolder(path, fs, outData,length+info.location.length());

		}else{
			FSDataInputStream intemp=null;
			long len=fstatus.getLen();
			try {
				outData.writeBoolean(false);//sigle files\
				outData.writeLong(len);
				outData.writeUTF(path.getName());
				intemp=fs.open(path);
				IOUtils.copyStream(intemp,outData,4096,len,null);

			}
			finally{
				IOUtils.close(intemp);
			}
		}
		outData.writeLong(-1);
		return true;
	}
	//read session id
	//read check code
	//if
	////error close
	//else 
	////read distance
	//////if(o) colse
	//////else(r)
	//////else(w)
	////if(ok) write(true)
	////else  close

	public void run(){
		String checkCode=null;
		while(true){
			if(socket==null||socket.isClosed()||!socket.isConnected()){
				System.out.println("socket is closed");
				this.closeConnect();
				return;
			}
			DataInputStream inData;
			DataOutputStream outData;
			try {
				inData = new DataInputStream(socket.getInputStream());
				outData = new DataOutputStream(socket.getOutputStream());
			} catch (IOException e) {
				System.out.println(e.getMessage());
				e.printStackTrace();
				this.closeConnect();
				return;
			}
			String sessionId;
			try {
				sessionId = inData.readUTF();
			} catch (IOException e1) {
				this.closeConnect();
				System.out.println("can not get session id.");
				return;
			}

			//if(state==null){
			state=LaUDDataServer.authorizedUsers.get(sessionId);
			//}
			if(state==null){
				this.closeConnect();
				logger.debug("wrong sessionId,can not get user session.");
				return ;
			}

			try {
				checkCode = inData.readUTF();
			} catch (IOException e1) {
				this.closeConnect();
				state.getFileDataSchedule().clear();
				LaUDDataServer.authorizedUsers.remove(sessionId);
				logger.debug("can not get checkcode .");
				return;
			}
			if(checkCode.equals("")){
				try {
					closeFs();
					outData.writeBoolean(true);
				} catch (IOException e) {
					e.printStackTrace();
				}finally{
					this.closeConnect();
				}
				return;

			}
			info=state.getFileInfo(checkCode);
			if(info==null){
				this.closeConnect();
				logger.debug("wrong checkCode,not allowed user to transfer files.");
				return ;
			}
			//state.setDataSocket(socket);
			char dis=LaUDJdbcUtils.OTHERS;
			try {
				dis = inData.readChar();
			} catch (IOException e1) {
				this.closeConnect();
				logger.debug(e1.getMessage());
				return;
			}
			try {
				if(dis==LaUDJdbcUtils.OTHERS){
					//cn.edu.thu.laud.ls.client send a wrong local filePath,do nothing
					logger.debug("nothing to tranfer.");
					outData.writeBoolean(true);//"go on" command
					continue;
				}
				boolean isok=false;
				try {
					fs=resetFS(info.getType());
				} catch (IOException e1) {
					e1.printStackTrace();
					this.closeConnect();
					logger.debug(e1.getMessage());
					state.addFileDataScheduleResult(checkCode, e1.getMessage());
				}

				if(dis==LaUDJdbcUtils.WRITE_INTO &&info.getDirection()==LaUDJdbcUtils.WRITE_INTO){//write into LaUD cloud.
					logger.debug("----------write into Laud cloud");
					isok=this.writeIntoLaud(inData, outData);
				}
				else if (dis==LaUDJdbcUtils.READ_FROM &&info.getDirection()==LaUDJdbcUtils.READ_FROM){//read from LaUD cloud.
					logger.debug("----------read from Laud cloud");
					isok=this.readFromLaud(inData,outData);
				}else{
					logger.debug("wrong transfer direction.");
					this.closeConnect();
					return;
				}
				outData.writeBoolean(true);//"go on" command
			} catch (Exception e) {
				e.printStackTrace();
				state.addFileDataScheduleResult(checkCode, e.getMessage());
				this.closeConnect();
			} finally{
				if(state!=null&&checkCode!=null){
					state.removeFileDataSchedule(checkCode);
					checkCode=null;
				}
			}
		}
	}



	//mendeley

	private FileSystem resetFS(FileStoreType fileStoreType) throws IOException {
		try{
			if (this.fs==null){
				conf.set("fs.default.name", fileStoreType.equals(FileStoreType.HDFS)?HDFS_PATH:CASSANDRA_PATH);
				fs = FileSystem.get(conf);
			}
			//TODO we need not reConnect a new Fs every time, but we can not find fs.isClosed() function
			else if(!this.fsType.equals(fileStoreType)){
				fs.close();
				//			Configuration conf = new Configuration();
				conf.set("fs.default.name", fileStoreType.equals(FileStoreType.HDFS)?HDFS_PATH:CASSANDRA_PATH);
				fs = FileSystem.get(conf);
			}
		}catch(Exception e){
			e.printStackTrace();
			throw new IOException(e);
		}catch(Throwable e){
			e.printStackTrace();
		}
		return fs;
	}


	private void closeConnect(){
		logger.debug("who close me????????????????????");

		IOUtils.close(socket);
		if(state!=null){
			state.setLocked(false);
		}
		closeFs();
	}
	private void closeFs(){
		if(fs!=null){
			try {
				fs.close();
			} catch (IOException e) {
				e.printStackTrace();
			}finally{
				fs=null;
			}
		}
	}
}
